package com.zhangwoo.analyser.jobs.matcher;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

class AllDataAnalyserMapper extends MapReduceBase implements
		Mapper<LongWritable, Text, Text, Text> {
	JobConf job = null;

	@Override
	public void configure(JobConf job) {
		this.job = job;
	}

	public void map(LongWritable key, Text value,
			OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
//		new TextMatcherWithKPE().processText(job.get("strategy"),
//				job.get("good"), job.get("bad"), value.toString());
	}
}

class AllDataAnalyserReducer extends MapReduceBase implements
		Reducer<Text, Text, Text, Text> {

	public void reduce(Text key, Iterator<Text> values,
			OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
	}
}

public class AllDataAnalyser {
	/**
	 * @param args
	 * @throws IOException
	 */
	public static void main(String[] args) throws IOException {
		if (args.length == 0)
			System.exit(-1);

		JobConf conf = new JobConf(AllDataAnalyser.class);
		conf.setJobName("Analyser_of_AllData@" + new java.util.Date().getTime());
		conf.set("strategy", args[0]);
		conf.set("good", args[1]);
		conf.set("bad", args[2]);
		for (int i = 3; i < args.length; i++) {
			FileInputFormat.addInputPaths(conf, args[i]);
		}
		FileOutputFormat.setOutputPath(conf, new Path("result/"
				+ new java.util.Date().getTime()));

		conf.setMapperClass(AllDataAnalyserMapper.class);
		conf.setReducerClass(AllDataAnalyserReducer.class);
		conf.setNumReduceTasks(0);

		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		JobClient.runJob(conf);
	}
}